package mrsimulator

import actors.{TIMEOUT, Actor}
import collection.JavaConverters._
import java.util.List
import java.util.LinkedList
import collection.mutable
import log.PVLogger

/**
 * Created with IntelliJ IDEA.
 * User: codefish
 * Date: 11/7/12
 * Time: 10:18 PM
 * To change this template use File | Settings | File Templates.
 */

object Stop
object Reduce
case class ShuffleDone(reduceId: String)
case class TaskDone(task: Task)

class TaskTrackerSimulator (vid: String, jt: HeartbeatMonitor) extends Actor{
    val vNodeId = vid
    val jobTracker = jt
    val numOfMapSlot = 2
    val numOfReduceSlot = 1
    val heartbeatInterval = 1000.toLong

    var numOfRunningMap = 0
    var numOfRunningReduce = 0

    private[TaskTrackerSimulator] var finishedMaps = new LinkedList[Task]()
    private[TaskTrackerSimulator] var finishedReduces = new LinkedList[Task]()

    private [TaskTrackerSimulator] var runningReduces = mutable.HashMap[String, ReduceTaskInProgress]()
    val LOG = PVLogger.getLogger(this.getClass)

    def getVNodeId(): String = {
        return vNodeId
    }

    def act(){
        var lastTick = System.currentTimeMillis()
        var waitTime = heartbeatInterval
        loop{
            reactWithin(waitTime){
                case TIMEOUT =>
                    LOG.debug("timeout")
                    val action = jobTracker.heartbeat(this.prepare())
                    val newMaps = action.getMapTaskToLaunch
                    var itr = newMaps.iterator()
                    while(itr.hasNext){
                        launchMapTask(itr.next)
                    }
                    val newReduces = action.getReduceTaskToLaunch
                    itr = newReduces.iterator()
                    while(itr.hasNext){
                        launchReduceTask(itr.next)
                    }
                    //TODO-FIXME: get reduce to notify
                    val notificationToReduce = action.getReduceTaskToNotify
                    notifyReduce(notificationToReduce)
                    resetClock()
                case TaskDone(task) =>
                    if (sender.isInstanceOf[MapTaskInProgress]){
                        LOG.debug("map " + task.getTaskId + " done")
                        numOfRunningMap -= 1
                        finishedMaps.add(task)
                    } else {
                        LOG.debug("reduce " + task.getTaskId + " done")
                        numOfRunningReduce -= 1
                        this.runningReduces.remove(task.getTaskId)
                        finishedReduces.add(task)
                    }
                    setClock()
                case ShuffleDone(reduceId) =>
                    LOG.debug("shuffledone")
                    // Shuffle of this task is done, start to run the real reduce function
                    this.runningReduces.get(reduceId).get ! Reduce
                    setClock()
                case Stop =>
                    LOG.debug("exit")
                    exit
            }
        }
        def setClock(){
            waitTime = heartbeatInterval-(System.currentTimeMillis()-lastTick)
            if (waitTime < 0) waitTime = 1.toLong
        }
        def resetClock(){
            lastTick = System.currentTimeMillis()
            waitTime = heartbeatInterval
        }
    }

    def notifyReduce(receivers: List[Task]){
        val itr = receivers.iterator
        while (itr.hasNext){
            val next = itr.next
            this.runningReduces.get(next.getTaskId).get ! Reduce
        }
    }

    def launchMapTask(map: Task){
        numOfRunningMap += 1
        new MapTaskInProgress(map, this).start
        LOG.debug("map " + map.getTaskId + " launched")
    }

    def launchReduceTask(reduce: Task){
        numOfRunningReduce += 1
        val newReduce = new ReduceTaskInProgress(reduce, this)
        runningReduces.put(reduce.getTaskId, newReduce)
        newReduce.start
        LOG.debug("reduce " + reduce.getTaskId + " launched")
    }

    def prepare(): TaskTrackerStatus = {
        val status  = new TaskTrackerStatus(numOfMapSlot-numOfRunningMap,
                                numOfReduceSlot-numOfRunningReduce, vNodeId, this)
        status.setMapFinishReport(finishedMaps)
        status.setReduceFinishReport(finishedReduces)
        status.setRunningReduceTasks(runningReduces.values.toList.map(x=>x.task).asJava)
        finishedMaps = new LinkedList[Task]()
        finishedReduces = new LinkedList[Task]()
        return status
    }

    def stop(){
        this ! Stop
    }

    def startReduce(id: String){
        LOG.debug("Signaling running reduce " + id)
        this ! new ShuffleDone(id)
    }
}

